package com.gxlevi.task

import com.gxlevi.bean.ClickLogWide
import com.gxlevi.util.HBaseUtil
import org.apache.flink.streaming.api.scala.{DataStream, KeyedStream, WindowedStream}
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.api.scala._

case class ChannelNetWork(
                           var channelId: String,
                           var network: String,
                           var date: String,
                           var pv: Long,
                           var uv: Long,
                           var newCount: Long,
                           var oldCount: Long
                         )

object ChannelNetWorkTask extends BaseTask[ChannelNetWork] {
  override def map(clickWideDataStream: DataStream[ClickLogWide]): DataStream[ChannelNetWork] = {
    clickWideDataStream.flatMap {
      clickLogWide =>
        List(
          ChannelNetWork(
            clickLogWide.channelID,
            clickLogWide.network,
            clickLogWide.yearMonth,
            clickLogWide.count,
            clickLogWide.isMonthNew,
            clickLogWide.isNew,
            isOld(clickLogWide.isNew, clickLogWide.isMonthNew)
          ),
          ChannelNetWork(
            clickLogWide.channelID,
            clickLogWide.network,
            clickLogWide.yearMonthDay,
            clickLogWide.count,
            clickLogWide.isDayNew,
            clickLogWide.isNew,
            isOld(clickLogWide.isNew, clickLogWide.isDayNew)
          ),
          ChannelNetWork(
            clickLogWide.channelID,
            clickLogWide.network,
            clickLogWide.yearMonthDayHour,
            clickLogWide.count,
            clickLogWide.isHourNew,
            clickLogWide.isNew,
            isOld(clickLogWide.isNew, clickLogWide.isHourNew)
          )
        )
    }
  }

  override def keyBy(mapStream: DataStream[ChannelNetWork]): KeyedStream[ChannelNetWork, String] = {
    mapStream.keyBy {
      network =>
        network.channelId + ":" + network.network + ":" + network.date
    }
  }

  override def reduce(windowedStream: WindowedStream[ChannelNetWork, String, TimeWindow]): DataStream[ChannelNetWork] = {
    windowedStream.reduce {
      (t1, t2) =>
        ChannelNetWork(t1.channelId, t1.network, t1.date, t1.pv + t2.pv, t1.uv + t2.uv, t1.newCount + t2.newCount, t1.oldCount + t2.oldCount)
    }
  }

  override def sinkToHBase(reduceDataStream: DataStream[ChannelNetWork]): Unit = {
    reduceDataStream.addSink {
      network => {
        tableName = "channel_network"
        rowKey = s"${network.channelId}:${network.network}:${network.date}"
        val netWorkColumn = "network"
        val map = HBaseUtil.getMapData(tableName, rowKey, clfName, List(
          pvColumn, uvColumn, newCountColumn, oldCountColumn
        ))

        HBaseUtil.putMapData(tableName, rowKey, clfName, Map(
          channelIdColumn -> network.channelId,
          netWorkColumn -> network.network,
          dateColumn -> network.date,
          pvColumn -> getTotal(map, pvColumn, network.pv),
          uvColumn -> getTotal(map, uvColumn, network.uv),
          newCountColumn -> getTotal(map, newCountColumn, network.newCount),
          oldCountColumn -> getTotal(map, oldCountColumn, network.oldCount)
        ))
      }
    }
  }
}
